package com.nageoffer.shortlink.project.mq.consumer;

import com.nageoffer.shortlink.project.dto.biz.ShortLinkStatsRecordDTO;
import com.nageoffer.shortlink.project.service.ShortLinkService;
import lombok.RequiredArgsConstructor;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import java.util.concurrent.Executors;
import java.util.concurrent.locks.LockSupport;

import static com.nageoffer.shortlink.project.common.constant.RedisKeyConstant.DELAY_QUEUE_STATS_KEY;

/**
 * 延迟记录短链接统计组件
 */
@Component
@Deprecated
@RequiredArgsConstructor
public class DelayShortLinkStatsConsumer implements InitializingBean {

    private final RedissonClient redissonClient;
    private final ShortLinkService shortLinkService;

    public void onMessage() {
        // 4. 创建单线程执行器，专门用于处理延迟队列消息
        Executors.newSingleThreadExecutor(
                runnable -> {
                    Thread thread = new Thread(runnable);
                    // 5. 设置线程名称，便于监控和调试
                    thread.setName("delay_short-link_stats_consumer");
                    // 6. 设置为守护线程，JVM退出时不会阻塞
                    thread.setDaemon(Boolean.TRUE);
                    return thread;
                })
                .execute(() -> {
                    // 7. 获取与生产者相同的队列实例
                    RBlockingDeque<ShortLinkStatsRecordDTO> blockingDeque = redissonClient.getBlockingDeque(DELAY_QUEUE_STATS_KEY);
                    RDelayedQueue<ShortLinkStatsRecordDTO> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);

                    // 8. 无限循环，持续消费队列中的消息
                    for (; ; ) {
                        try {
                            // 9. 非阻塞方式获取队列中的消息（延迟时间到达的消息）
                            ShortLinkStatsRecordDTO statsRecord = delayedQueue.poll();

                            // 10. 如果获取到消息，则处理统计逻辑
                            if (statsRecord != null) {
                                // 11. 调用短链接服务进行统计处理，传入null表示让服务内部自行获取gid
                                shortLinkService.shortLinkStats(statsRecord);
                                continue; // 继续处理下一条消息
                            }

                            // 12. 如果没有消息，线程休眠500毫秒，避免CPU空转
                            // LockSupport.parkUntil使用绝对时间，这里实际是相对当前时间500ms
                            LockSupport.parkUntil(500);

                        } catch (Throwable ignored) {
                            // 13. 忽略所有异常，确保消费线程不会因异常而终止
                            // 这样设计保证了消费者的高可用性
                        }
                    }
                });
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        // 14. Spring Bean初始化完成后自动启动消费者
        // 实现InitializingBean接口，确保容器启动时就开始消费
        //onMessage();
    }
}
